package com.ry.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.ry.bean.People;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.websocket.SendResult;

@RestController
@RequestMapping("/kafka")
public class KafkaProducerConsumer {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    //普通发送
    @GetMapping("/normal/{message}")
    public void sendMessage(@PathVariable String message){
        kafkaTemplate.send("topic1", message);
    }

    //增加发送结果的回调函数
    @GetMapping("/callback/{message}")
    public void sendMessageCallback(@PathVariable String message){
        kafkaTemplate.send("topic1", message).addCallback(success -> {
            RecordMetadata record = success.getRecordMetadata();
            System.out.println("消息发送成功 --- 主题："+record.topic() +" 分区："+record.partition()+" 偏移量："+record.offset());
        }, failure ->{
            System.out.println("消息发送失败 --- "+failure.getMessage());
        });
    }

    //发送对象
    @GetMapping("/object/{message}")
    public void sendMessageObject(@PathVariable String message){
        People p = new People();
        p.setId(1);
        p.setName(message);
        p.setAge(10);
        p.setCustomerId(10000);
        kafkaTemplate.send("topic2", JSON.toJSONString(p));
    }


}
